package cn.fan.mqtt.server;
 
 

import cn.fan.mqtt.server.util.RemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.beans.factory.annotation.Value;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;


public class BootNettyServer {



	private int port = 1883;
	private static final  int  CPU =Runtime.getRuntime().availableProcessors();
	private EventLoopGroup bossGroup;
 
	private EventLoopGroup workGroup;
 
	/**
	 * 	启动服务
	 * @throws InterruptedException 
	 */
	public void startup() throws InterruptedException {
 
		try {
//			bossGroup = new NioEventLoopGroup(1);
//			workGroup = new NioEventLoopGroup();
			initEventPool();
			ServerBootstrap bootstrap = new ServerBootstrap();
			bootstrap.group(bossGroup, workGroup);
			bootstrap.channel(NioServerSocketChannel.class);
 
			bootstrap.option(ChannelOption.SO_REUSEADDR, true)
					.option(ChannelOption.SO_BACKLOG, 1024)
					.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
					.option(ChannelOption.SO_RCVBUF, 10485760);
 
			bootstrap.childOption(ChannelOption.TCP_NODELAY, true)
					.childOption(ChannelOption.SO_KEEPALIVE, true)
					.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
// 这ChannelInitializer全局是单例可以做一些统计的事
			bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
						@Override
						protected void initChannel(SocketChannel ch) {
							ChannelPipeline channelPipeline = ch.pipeline();
							// 设置读写空闲超时时间
							channelPipeline.addLast(new IdleStateHandler(600, 600, 1200));
							channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
							channelPipeline.addLast("decoder", new MqttDecoder());
							channelPipeline.addLast(new BootChannelInboundHandler());
						}
					});
			ChannelFuture f = bootstrap.bind(port).sync();
			f.channel().closeFuture().sync();
			
		} catch (Exception e) {
			System.out.println("start exception"+e.toString());
		}finally {
			shutdown();
		}
 
	}
	/**
	 * 初始化EnentPool 参数
	 */
	private void  initEventPool(){

		if(useEpoll()){
			bossGroup = new EpollEventLoopGroup(CPU, new ThreadFactory() {
				private AtomicInteger index = new AtomicInteger(0);

				@Override
				public Thread newThread(Runnable r) {
					return new Thread(r, "LINUX_BOSS_" + index.incrementAndGet());
				}
			});
			workGroup = new EpollEventLoopGroup(CPU*2, new ThreadFactory() {
				private AtomicInteger index = new AtomicInteger(0);

				@Override
				public Thread newThread(Runnable r) {
					return new Thread(r, "LINUX_WORK_" + index.incrementAndGet());
				}
			});

		}
		else {
			bossGroup = new NioEventLoopGroup(CPU, new ThreadFactory() {
				private AtomicInteger index = new AtomicInteger(0);

				@Override
				public Thread newThread(Runnable r) {
					return new Thread(r, "BOSS_" + index.incrementAndGet());
				}
			});
			workGroup = new NioEventLoopGroup(CPU*2, new ThreadFactory() {
				private AtomicInteger index = new AtomicInteger(0);

				@Override
				public Thread newThread(Runnable r) {
					return new Thread(r, "WORK_" + index.incrementAndGet());
				}
			});
		}
	}
	/**
	 * 	关闭服务
	 */
	public void shutdown() throws InterruptedException {
		if (workGroup != null && bossGroup != null) {
			bossGroup.shutdownGracefully();
			workGroup.shutdownGracefully();
			System.out.println("shutdown success");
		}
	}
	private boolean useEpoll() {
		return RemotingUtil.isLinuxPlatform()
				&& Epoll.isAvailable();
	}
}